package drds.global_transaction.api.rpc.rpc_server;

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeoutException;

import drds.global_transaction.api.api.RpcContext;
import drds.global_transaction.api.in_bound_handler.TransactionMessageHandler;
import drds.global_transaction.api.message_codec.HeartbeatMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.RegisterResourceManagerRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.RegisterTransactionManagerRequestMessage;
import drds.global_transaction.api.message_codec.rpc_message.RpcMessage;
import drds.global_transaction.api.rpc.config.ServerConfig;
import drds.global_transaction.common.exception.FrameworkErrorCode;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Sharable
public class RpcServer extends AbstractRpcServer implements Sender
{
	@Setter
	@Getter
	protected MessageListener messageListener;
	@Setter
	@Getter
	private TransactionMessageHandler transactionMessageHandler;
	@Setter
	@Getter
	private RegisterCheckAuthHandler registerCheckAuthHandler;

	public RpcServer(ThreadPoolExecutor messageExecutor)
	{
		super(new ServerConfig(), messageExecutor);
	}

	public void setHandler(TransactionMessageHandler transactionMessageHandler)
	{
		setHandler(transactionMessageHandler, null);
	}

	public void setHandler(TransactionMessageHandler transactionMessageHandler, RegisterCheckAuthHandler registerCheckAuthHandler)
	{
		this.transactionMessageHandler = transactionMessageHandler;
		this.registerCheckAuthHandler = registerCheckAuthHandler;
	}

	@Override
	public void init()
	{
		super.init();
		setChannelHandlers(new ChannelHandler[] { RpcServer.this });
		MessageListenerImpl messageListener = new MessageListenerImpl(transactionMessageHandler);
		messageListener.setSender(this);
		this.setMessageListener(messageListener);
		super.start();

	}

	private void closeChannelHandlerContext(ChannelHandlerContext channelHandlerContext)
	{
		if (log.isInfoEnabled())
		{
			log.info("closeChannelHandlerContext channel:" + channelHandlerContext.channel());
		}
		channelHandlerContext.disconnect();
		channelHandlerContext.close();
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext channelHandlerContext, Object event)
	{
		if (event instanceof IdleStateEvent)
		{
			IdleStateEvent idleStateEvent = (IdleStateEvent) event;
			if (idleStateEvent.state() == IdleState.READER_IDLE)
			{
				rpcContextRelease(channelHandlerContext);
				try
				{
					closeChannelHandlerContext(channelHandlerContext);
				} catch (Exception e)
				{
					log.error(e.getMessage());
				}
			}
		}
	}

	@Override
	public void destroy()
	{
		super.destroy();
		super.shutdown();
		if (log.isInfoEnabled())
		{
			log.info("destroyed rpcServer");
		}
	}

	/**
	 * Send request. handle async branch globalTransactionCommit and
	 * globalTransactionRollback
	 *
	 * @param resourceId    the db key
	 * @param clientIp      the client ip
	 * @param applicationId the client app name
	 * @param request       the msg
	 */
	@Override
	public void sendRequest(String resourceId, String clientIp, String applicationId, Object request)
	{
		Channel channel = ChannelManager.getChannel(resourceId, applicationId, clientIp);
		if (channel == null)
		{
			throw new FrameworkException("resource_manager_client client is not connected. dbkey:" + resourceId + ",clientIp:" + clientIp);
		} else
		{
			try
			{
				super.sendRequest(channel, request);// 客户端channel
			} catch (FrameworkException e)
			{
				if (e.getFrameworkErrorCode() == FrameworkErrorCode.ChannelIsNotWritable)
				{
					ChannelManager.releaseRpcContext(channel);
					throw e;
				}
			}
		}
	}

	/**
	 * Send response. redress,merge msg
	 */
	@Override
	public void sendResponse(long messageId, String resourceId, String clientIp, String applicationId, Object msg)
	{
		Channel clientChannel = ChannelManager.getChannel(resourceId, applicationId, clientIp);
		if (clientChannel != null)
		{
			try
			{
				super.sendResponse(messageId, clientChannel, msg);
			} catch (FrameworkException e)
			{
				if (e.getFrameworkErrorCode() == FrameworkErrorCode.ChannelIsNotWritable)
				{
					log.error("channel is not writeable,channel:" + clientChannel);
					ChannelManager.releaseRpcContext(clientChannel);
				}
				throw e;
			}
		} else
		{
			throw new RuntimeException("channel is error. channel:" + clientChannel);
		}
	}

	/**
	 * Send response. resource_manager_client reg,rpc reg,inner response
	 *
	 * @param channel   the channel
	 * @param messageId the msg id
	 * @param message   the msg
	 */
	@Override
	public void sendResponse(Channel channel, long messageId, Object message)
	{
		Channel clientChannel = channel;
		if (!(message instanceof HeartbeatMessage))
		{
			clientChannel = ChannelManager.getChannel(channel);
		}
		if (clientChannel != null)
		{
			super.sendResponse(messageId, clientChannel, message);
		} else
		{
			throw new RuntimeException("channel is error. channel:" + clientChannel);
		}
	}

	/**
	 * Send request with response object. send syn request for
	 * resource_manager_client
	 *
	 * @param resourceId    the db key
	 * @param clientIp      the client ip
	 * @param applicationId the client app name
	 * @param msg           the msg
	 * @param timeout       the timeout
	 * @return the object
	 * @throws TimeoutException the timeout exception
	 */
	@Override
	public Object sendSynRequest(String resourceId, String clientIp, String applicationId, Object msg, long timeout) throws TimeoutException
	{
		Channel clientChannel = ChannelManager.getChannel(resourceId, applicationId, clientIp);
		if (clientChannel != null)
		{
			return super.sendAsynchronousRequestWithResponse(null, clientChannel, msg, timeout);
		} else
		{
			throw new RuntimeException("resource_manager_client client is not connected. dbkey:" + resourceId + ",clientIp:" + clientIp);
		}
	}

	@Override
	public Object sendSynRequest(String resourceId, String clientIp, String applicationId, Object msg) throws TimeoutException
	{
		return sendSynRequest(resourceId, clientIp, applicationId, msg, ServerConfig.getRpcRequestTimeout());
	}

	@Override
	public void dispatch(long messageId, ChannelHandlerContext channelHandlerContext, Object message)
	{
		if (message instanceof RegisterResourceManagerRequestMessage)
		{
			messageListener.onRegisterResourceManagerRequest(channelHandlerContext, this, messageId, (RegisterResourceManagerRequestMessage) message, registerCheckAuthHandler);
		} else
		{
			if (ChannelManager.isRegistered(channelHandlerContext.channel()))
			{
				messageListener.onTransactionMessage(channelHandlerContext, this, messageId, message);
			} else
			{
				try
				{
					closeChannelHandlerContext(channelHandlerContext);
				} catch (Exception exx)
				{
					log.error(exx.getMessage());
				}
				if (log.isInfoEnabled())
				{
					log.info(String.format("close a unhandled connection! [%s]", channelHandlerContext.channel().toString()));
				}
			}
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception
	{
		if (threadPoolExecutor.isShutdown())
		{
			return;
		}
		rpcContextRelease(channelHandlerContext);
		super.channelInactive(channelHandlerContext);
	}

	private void rpcContextRelease(ChannelHandlerContext channelHandlerContext)
	{
		final String ipAndPort = Nets.getHostAddressAndPort(channelHandlerContext.channel().remoteAddress());
		RpcContext rpcContext = ChannelManager.getRpcContext(channelHandlerContext.channel());
		if (log.isInfoEnabled())
		{
			log.info(ipAndPort + " to transaction_coordinator channel inactive.");
		}
		if (null != rpcContext && null != rpcContext.getRole())
		{
			rpcContext.release();// rpcContext release
			if (log.isInfoEnabled())
			{
				log.info("remove channel:" + channelHandlerContext.channel() + "global_transaction_id_context:" + rpcContext);
			}
		} else
		{
			if (log.isInfoEnabled())
			{
				log.info("remove unused channel:" + channelHandlerContext.channel());
			}
		}
	}

	@Override
	public void channelRead(final ChannelHandlerContext channelHandlerContext, Object message) throws Exception
	{
		if (message instanceof RpcMessage)
		{
			RpcMessage rpcMessage = (RpcMessage) message;
			if (rpcMessage.getMessageCodec() instanceof RegisterTransactionManagerRequestMessage)
			{
				RegisterTransactionManagerRequestMessage registerTransactionManagerRequestMessage = (RegisterTransactionManagerRequestMessage) rpcMessage.getMessageCodec();
				messageListener.onRegisterTransactionManagerRequest(channelHandlerContext, this, rpcMessage.getId(), registerTransactionManagerRequestMessage, registerCheckAuthHandler);
				return;
			}
			if (rpcMessage.getMessageCodec() == HeartbeatMessage.ping)
			{
				messageListener.onCheckMessageRequest(channelHandlerContext, this, rpcMessage.getId());
				return;
			}
		}
		super.channelRead(channelHandlerContext, message);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable cause) throws Exception
	{
		if (log.isInfoEnabled())
		{
			log.info("channel exx:" + cause.getMessage() + ",channel:" + channelHandlerContext.channel());
		}
		ChannelManager.releaseRpcContext(channelHandlerContext.channel());
		super.exceptionCaught(channelHandlerContext, cause);
	}
}
